package websocket

import (
	"errors"
	"github.com/gorilla/websocket"
	"github.com/sirupsen/logrus"
	"net/http"
	"sync"
	"time"
)

var (
	ErrWebsocketClosed = errors.New("websocket closed")
)

// 客户端读写消息
type (
	WsClient struct {
		url           string
		requestHeader http.Header

		wsSocket *websocket.Conn // 底层websocket
		inChan   chan WsMessage  // 读队列
		outChan  chan WsMessage  // 写队列

		closeChan       chan struct{} //关闭通知
		reconnectChan   chan struct{} //重连消息通知
		connectSuccess  chan struct{} //连接成功，这个chan会被关闭
		reconnectNotify chan struct{} //连接出错通知，需要重连
		connecting      bool          //连接中标识
		mux             sync.Mutex    //避免同一时间发送多个重连通知
		onceClose       sync.Once     // 避免重复关闭管道

		ConnectionIntervalSecond int //重连间隔时间
		PingIntervalSecond       int //ping间隔时间 Send pings to peer with this period. Must be less than pongWait.
	}

	WsMessage struct {
		MessageType int
		Data        []byte
	}
)

func NewWsClient(url string, requestHeader http.Header) *WsClient {
	return &WsClient{
		url:                      url,
		requestHeader:            requestHeader,
		inChan:                   make(chan WsMessage),
		outChan:                  make(chan WsMessage),
		onceClose:                sync.Once{},
		closeChan:                make(chan struct{}),
		reconnectChan:            make(chan struct{}),
		connectSuccess:           make(chan struct{}),
		reconnectNotify:          make(chan struct{}),
		connecting:               false,
		mux:                      sync.Mutex{},
		ConnectionIntervalSecond: 10,
		PingIntervalSecond:       30,
	}
}

func (wc *WsClient) Run() {
	//连接
	wc.connect()
	//重连
	go wc.reconnect()
	//心跳
	go wc.ping()
	//读
	go wc.readLoop()
	//写
	go wc.writeLoop()

}

// Read 读消息
func (wc *WsClient) Read() (msg WsMessage, err error) {
	select {
	case msg = <-wc.inChan:
		return msg, nil
	case <-wc.closeChan:
	}

	err = ErrWebsocketClosed
	return
}

// Write 写消息
func (wc *WsClient) Write(msg []byte, msgType ...int) (err error) {
	ms := WsMessage{Data: msg, MessageType: websocket.TextMessage}
	if len(msgType) > 0 {
		ms.MessageType = msgType[0]
	}

	select {
	case wc.outChan <- ms:
		return
	case <-wc.closeChan:
	}

	err = ErrWebsocketClosed
	return
}

// Close 关闭
func (wc *WsClient) Close() {
	wc.onceClose.Do(func() {
		if wc.wsSocket != nil {
			_ = wc.wsSocket.Close()
		}
		close(wc.closeChan)
	})
}

// reconnect 重连
func (wc *WsClient) reconnect() {
	for {
		select {
		case <-wc.reconnectNotify:
			// 未连接成功，这里回卡住
			if wc.connecting {
				logrus.Infof("Ws client reconnect skip")
				continue
			}

			logrus.Infof("Ws client reconnecting")
			wc.connect()
		case <-wc.closeChan:
			return
		}
	}

}

// connect 连接
func (wc *WsClient) connect() {
	var (
		err error
	)

	wc.connectSuccess = make(chan struct{})
	wc.connecting = true

	defer func() {
		close(wc.connectSuccess)
		wc.connecting = false
	}()
	logrus.Infof("WsClient connecting")

	for {
		select {
		case <-wc.closeChan:
			return
		default:
			if wc.wsSocket, _, err = websocket.DefaultDialer.Dial("ws://"+wc.url, wc.requestHeader); err == nil {
				logrus.Infof("WsClient connect success")
				return
			}

			logrus.Infof("WsClient connect err:%s,reconnect in 10 seconds", err)

			time.Sleep(time.Duration(wc.ConnectionIntervalSecond) * time.Second)

		}
	}

}

// readLoop 从socket中将消息读到chan
func (wc *WsClient) readLoop() {
	var (
		err     error
		msgType int
		data    []byte
	)
	for {
		<-wc.connectSuccess

		// 读一个message
		if msgType, data, err = wc.wsSocket.ReadMessage(); err != nil {
			logrus.Infof("WsClient readLoop err:%s", err)
			wc.notifyReconnect()
			continue
		}
		msg := WsMessage{
			msgType,
			data,
		}
		// 放入请求队列
		select {
		case wc.inChan <- msg:
		case <-wc.closeChan:
			return
		}
	}
}

// writeLoop 将chan中消息写入chan
func (wc *WsClient) writeLoop() {
	var (
		err error
	)
	for {
		select {
		// 取一个应答
		case msg := <-wc.outChan:
			//连接不可用，会卡住
			<-wc.connectSuccess
			// 写给websocket
			if err = wc.wsSocket.WriteMessage(msg.MessageType, msg.Data); err != nil {
				logrus.Infof("WsClient writeLoop err:%s", err)
				//写消息错误，通知重连
				wc.notifyReconnect()
			}
		case <-wc.closeChan:
			return
		}
	}
}

// ping 连接保持
func (wc *WsClient) ping() {
	ticker := time.NewTicker(time.Duration(wc.PingIntervalSecond) * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:

			// 连接不可用，这里会卡住
			<-wc.connectSuccess

			if err := wc.wsSocket.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(10*time.Second)); err != nil {
				logrus.Infof("WsClient ping err:%s", err)

				wc.notifyReconnect()
			}

		case <-wc.closeChan:
			return
		}
	}
}

// notifyReconnect 通知重连
func (wc *WsClient) notifyReconnect() {
	if !wc.mux.TryLock() {
		return
	}
	wc.reconnectNotify <- struct{}{}
	wc.mux.Unlock()
}
